package com.dongdongshop.mq;


import cn.hutool.json.JSONObject;
import cn.hutool.json.JSONUtil;
import com.dongdongshop.aldy.ExamPhone;
import com.dongdongshop.pojo.TbUser;
import com.dongdongshop.service.TbOrderService;
import com.dongdongshop.service.TbUserService;
import org.apache.dubbo.config.annotation.DubboReference;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.apache.rocketmq.spring.core.RocketMQPushConsumerLifecycleListener;
import org.springframework.beans.factory.FactoryBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.List;

@Component
//消费者需要用注解配置topic主题 和消费者的组
@RocketMQMessageListener(consumerGroup = "2204dk-consumer",topic = "tx-order-score")
//消费者需要继承接口RocketMQListener<>可以通过泛型接受消费消息
//ack重试机制要继承RocketMQPushConsumerLifecycleListener
public class ConsumerOrderCred implements RocketMQListener<String>, RocketMQPushConsumerLifecycleListener {

    @Autowired
    private TbUserService userService;

    @DubboReference(check = false)
    private TbOrderService tbOrderService;
    //继承RocketMQListener的方法
    @Override
    public void onMessage(String s) { //继承RocketMQPushConsumerLifecycleListener的接口后这个方法就作废了
        System.out.println(s);
    }
    //继承RocketMQPushConsumerLifecycleListener后的方法
    @Override
    public void prepareStart(DefaultMQPushConsumer defaultMQPushConsumer) {


        //通过方法创建内部类内部类中写业务
        defaultMQPushConsumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            //list就是消费消息
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
                //单条消息直接下表0获取对象,可以通过对象获取更多属性
                MessageExt message = list.get(0);
                //new string接受消息内容 如果不转string会是一个二进制字符串b
                String msg = new String(message.getBody());
                JSONObject jsonObject = JSONUtil.parseObj(msg);
                String s = (String) jsonObject.get("money");
                String out_trade_no = (String) jsonObject.get("out_trade_no");
                String trade_no = (String) jsonObject.get("trade_no");

                //获取消息重试次数
                int reconsumeTimes = message.getReconsumeTimes();
                //ctrl+alt+t  生成try
                try {
                    String userid = tbOrderService.selectUserIdByOrderNumandTradeNum(out_trade_no,trade_no);
                    TbUser user = userService.selectByUsername(userid);
                    userService.orderScore(user.getId(),s);

                } catch (Exception e) {
                    e.printStackTrace();
                    if (reconsumeTimes >= 3){//如果重试次数大于3,没必要继续尝试返回成功,并且将消息放入死信队列
                        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                    }
                        return ConsumeConcurrentlyStatus.RECONSUME_LATER;//发生异常返回重新消费
                    }
                    //消费成功返回success
                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;//返回消费成功
            }
        });
    }
}
